package internal

import (
	"fmt"
	"strings"
)

// SchemaSync 配置文件
type SchemaSync struct {
	Config   *Config
	SourceDb *MyDb
	DestDb   *MyDb
}

var syncInstance *SchemaSync

// NewSchemaSync 对一个配置进行同步
func NewSchemaSync(config *Config) (*SchemaSync, error) {
	Debug("Creating new schema sync instance")
	
	if syncInstance == nil {
		syncInstance = new(SchemaSync)
		syncInstance.Config = config
		
		sourceDb, err := NewMyDb(config.SourceDSN, config.Schemas[0], "source", config.SourceSSH, config)
		if err != nil {
			return nil, NewDatabaseError("failed to create source database connection", err)
		}
		syncInstance.SourceDb = sourceDb
		
		destDb, err := NewMyDb(config.DestDSN, config.Schemas[0], "dest", config.DestSSH, config)
		if err != nil {
			return nil, NewDatabaseError("failed to create destination database connection", err)
		}
		syncInstance.DestDb = destDb
	}
	
	return syncInstance, nil
}

// use dbName
func (sc *SchemaSync) UseDb(dbname string) error {
	Debug("Switching to database: %s", dbname)
	
	var err error
	if sc.SourceDb.DbName != dbname {
		sc.SourceDb.Close()
		syncInstance.SourceDb, err = NewMyDb(sc.Config.SourceDSN, dbname, "source", sc.Config.SourceSSH, sc.Config)
		if err != nil {
			return NewDatabaseError("failed to switch source database", err)
		}
	}
	if sc.DestDb.DbName != dbname {
		sc.DestDb.Close()
		syncInstance.DestDb, err = NewMyDb(sc.Config.DestDSN, dbname, "dest", sc.Config.DestSSH, sc.Config)
		if err != nil {
			return NewDatabaseError("failed to switch destination database", err)
		}
	}
	
	fmt.Printf("------------------------ db %s -------------------------\n", dbname)
	return nil
}

// GetNewTableNames 获取所有新增加的表名
func (sc *SchemaSync) GetNewTableNames() ([]string, error) {
	sourceTables, err := sc.SourceDb.GetTableNames()
	if err != nil {
		return nil, err
	}
	
	destTables, err := sc.DestDb.GetTableNames()
	if err != nil {
		return nil, err
	}

	var newTables []string
	for _, name := range sourceTables {
		if !inStringSlice(name, destTables) {
			newTables = append(newTables, name)
		}
	}
	return newTables, nil
}

// 合并源数据库和目标数据库的表名
func (sc *SchemaSync) GetTableNames() ([]string, error) {
	sourceTables, err := sc.SourceDb.GetTableNames()
	if err != nil {
		return nil, err
	}
	
	destTables, err := sc.DestDb.GetTableNames()
	if err != nil {
		return nil, err
	}
	
	var tables []string
	tables = append(tables, destTables...)
	for _, name := range sourceTables {
		if !inStringSlice(name, tables) {
			tables = append(tables, name)
		}
	}
	return tables, nil
}

// RemoveTableSchemaConfig 删除表创建引擎信息，编码信息，分区信息，已修复同步表结构遇到分区表异常退出问题，
// 对于分区表，只会同步字段，索引，主键，外键的变更
func RemoveTableSchemaConfig(schema string) string {
	return strings.Split(schema, "ENGINE")[0]
}

func (sc *SchemaSync) getAlterDataByTable(table string, cfg *Config) (*TableAlterData, error) {
	sSchema, err := sc.SourceDb.GetTableSchema(table)
	if err != nil {
		return nil, err
	}
	
	dSchema, err := sc.DestDb.GetTableSchema(table)
	if err != nil {
		return nil, err
	}
	
	return sc.getAlterDataBySchema(table, sSchema, dSchema, cfg), nil
}

func (sc *SchemaSync) getAlterDataBySchema(table string, sSchema string, dSchema string, cfg *Config) *TableAlterData {
	alter := new(TableAlterData)
	alter.Table = table
	alter.Type = alterTypeNo
	alter.SchemaDiff = newSchemaDiff(table, RemoveTableSchemaConfig(sSchema), RemoveTableSchemaConfig(dSchema))

	if sSchema == dSchema {
		return alter
	}
	if len(sSchema) == 0 {
		alter.Type = alterTypeDropTable
		alter.Comment = "源数据库不存在，删除目标数据库多余的表"
		alter.SQL = append(alter.SQL, fmt.Sprintf("drop table `%s`;", table))
		return alter
	}
	if len(dSchema) == 0 {
		alter.Type = alterTypeCreate
		alter.Comment = "目标数据库不存在，创建"
		alter.SQL = append(alter.SQL, fmtTableCreateSQL(sSchema)+";")
		return alter
	}

	diffLines := sc.getSchemaDiff(alter)
	if len(diffLines) == 0 {
		return alter
	}
	alter.Type = alterTypeAlter
	if cfg.SingleSchemaChange {
		for _, line := range diffLines {
			ns := fmt.Sprintf("ALTER TABLE `%s`\n%s;", table, line)
			alter.SQL = append(alter.SQL, ns)
		}
	} else {
		ns := fmt.Sprintf("ALTER TABLE `%s`\n%s;", table, strings.Join(diffLines, ",\n"))
		alter.SQL = append(alter.SQL, ns)
	}

	return alter
}

func (sc *SchemaSync) getSchemaDiff(alter *TableAlterData) []string {
	sourceMyS := alter.SchemaDiff.Source
	destMyS := alter.SchemaDiff.Dest
	var beforeFieldName string
	var alterLines []string
	var fieldCount int = 0
	// 比对字段
	for el := sourceMyS.Fields.Front(); el != nil; el = el.Next() {
		var alterSQL string
		if destDt, has := destMyS.Fields.Get(el.Key); has {
			if el.Value != destDt {
				alterSQL = fmt.Sprintf("CHANGE `%s` %s", el.Key, el.Value)
			}
			beforeFieldName = el.Key.(string)
		} else {
			if len(beforeFieldName) == 0 {
				if fieldCount == 0 {
					alterSQL = "ADD " + el.Value.(string) + " FIRST"
				} else {
					alterSQL = "ADD " + el.Value.(string)
				}
			} else {
				alterSQL = fmt.Sprintf("ADD %s AFTER `%s`", el.Value.(string), beforeFieldName)
			}
			beforeFieldName = el.Key.(string)
		}

		if len(alterSQL) != 0 {
			alterLines = append(alterLines, alterSQL)
		}
		fieldCount++
	}

	// 源库已经删除的字段
	if sc.Config.Drop {
		for _, name := range destMyS.Fields.Keys() {
			if _, has := sourceMyS.Fields.Get(name); !has {
				alterSQL := fmt.Sprintf("drop `%s`", name)
				alterLines = append(alterLines, alterSQL)
			}
		}
	}

	// 多余的字段暂不删除

	// 比对索引
	for indexName, idx := range sourceMyS.IndexAll {
		dIdx, has := destMyS.IndexAll[indexName]
		var alterSQLs []string
		if has {
			if idx.SQL != dIdx.SQL {
				alterSQLs = append(alterSQLs, idx.alterAddSQL(true)...)
			}
		} else {
			alterSQLs = append(alterSQLs, idx.alterAddSQL(false)...)
		}
		if len(alterSQLs) > 0 {
			alterLines = append(alterLines, alterSQLs...)
		}
	}

	// drop index
	if sc.Config.Drop {
		for indexName, dIdx := range destMyS.IndexAll {
			var dropSQL string
			if _, has := sourceMyS.IndexAll[indexName]; !has {
				dropSQL = dIdx.alterDropSQL()
			}

			if len(dropSQL) != 0 {
				alterLines = append(alterLines, dropSQL)
			}
		}
	}

	// 比对外键
	for foreignName, idx := range sourceMyS.ForeignAll {
		dIdx, has := destMyS.ForeignAll[foreignName]
		var alterSQLs []string
		if has {
			if idx.SQL != dIdx.SQL {
				alterSQLs = append(alterSQLs, idx.alterAddSQL(true)...)
			}
		} else {
			alterSQLs = append(alterSQLs, idx.alterAddSQL(false)...)
		}
		if len(alterSQLs) > 0 {
			alterLines = append(alterLines, alterSQLs...)
		}
	}

	// drop 外键
	if sc.Config.Drop {
		for foreignName, dIdx := range destMyS.ForeignAll {
			var dropSQL string
			if _, has := sourceMyS.ForeignAll[foreignName]; !has {
				dropSQL = dIdx.alterDropSQL()
			}
			if len(dropSQL) != 0 {
				alterLines = append(alterLines, dropSQL)
			}
		}
	}

	return alterLines
}

// SyncSQL4Dest sync schema change
func (sc *SchemaSync) SyncSQL4Dest(sqlStr string, sqls []string) error {
	sqlStr = strings.TrimSpace(sqlStr)
	if len(sqlStr) == 0 {
		return nil
	}
	t := newMyTimer()
	ret, err := sc.DestDb.Query(sqlStr)

	defer func() {
		if ret != nil {
			err := ret.Close()
			if err != nil {
				Error("Close result error: %v", err)
				return
			}
		}
	}()

	// how to enable allowMultiQueries?
	if err != nil && len(sqls) > 1 {
		tx, errTx := sc.DestDb.Db.Begin()
		if errTx == nil {
			for _, sql := range sqls {
				ret, err = tx.Query(sql)
				if err != nil {
					Error("Error query_one: [%s] %v", sql, err)
					break
				}
			}
			if err == nil {
				err = tx.Commit()
			} else {
				_ = tx.Rollback()
			}
		}
	}
	t.stop()
	if err != nil {
		Error("EXEC_SQL_FAILED: %v", err)
		return err
	}
	_, err = ret.Columns()
	return err
}

// check data change
func (sc *SchemaSync) CheckDiffData(cfg *Config) error {
	if len(cfg.TablesCompareData) == 0 {
		fmt.Println("# Tables to CompareData is empty")
		return nil
	}

	allTables, err := sc.SourceDb.GetTableNames()
	if err != nil {
		return NewDatabaseError("failed to get source table names", err)
	}
	
	dstTables, err := sc.DestDb.GetTableNames()
	if err != nil {
		return NewDatabaseError("failed to get destination table names", err)
	}

	dataDiffTables := []string{}

	for _, table := range allTables {
		if !cfg.CheckMatchCompareDataTables(table) {
			continue
		}
		// 目标库没有此表
		if !inStringSlice(table, dstTables) {
			dataDiffTables = append(dataDiffTables, table)
			continue
		}

		// 查询第一个表
		rows1, err := sc.SourceDb.Query(fmt.Sprintf("CHECKSUM TABLE `%s`", table))
		if err != nil {
			return NewDatabaseError("failed to checksum source table", err)
		}
		defer rows1.Close()

		rows2, err := sc.DestDb.Query(fmt.Sprintf("CHECKSUM TABLE `%s`", table))
		if err != nil {
			return NewDatabaseError("failed to checksum destination table", err)
		}
		defer rows2.Close()

		// 比较两个表的数据
		var tab1, tab2 string
		var c1, c2 int64

		rows1.Next()
		rows2.Next()
		rows1.Scan(&tab1, &c1)
		rows2.Scan(&tab2, &c2)

		if c1 != c2 {
			dataDiffTables = append(dataDiffTables, table)
			continue
		}
	}

	if len(dataDiffTables) == 0 {
		fmt.Println("# no data of tables in difference")
	} else {
		fmt.Println("# !!!!! data diff tables: ", strings.Join(dataDiffTables, ", "))
	}
	
	return nil
}

// check procedures
func CheckAlterProcedure(cfg *Config) error {
	sc, err := NewSchemaSync(cfg)
	if err != nil {
		return err
	}
	
	allProcedures, err := sc.SourceDb.GetProcedureNames()
	if err != nil {
		return NewDatabaseError("failed to get source procedure names", err)
	}

	for _, proceName := range allProcedures {
		srcProcedureStr, err := sc.SourceDb.GetProcedureSchema(proceName)
		if err != nil {
			return NewDatabaseError("failed to get source procedure schema", err)
		}

		destProcedureStr, err := sc.DestDb.GetProcedureSchema(proceName)
		if err != nil {
			return NewDatabaseError("failed to get destination procedure schema", err)
		}

		if srcProcedureStr != destProcedureStr {
			fmt.Printf("procedure %s has changed: \n", proceName)
			fmt.Printf("Source: %s\n", srcProcedureStr)
			fmt.Printf("Destination: %s\n", destProcedureStr)
		}
	}
	return nil
}

// SyncConditionalData 根据条件同步表数据
func (sc *SchemaSync) SyncConditionalData(cfg *Config) error {
	if !cfg.CheckHasConditionalSync() {
		Info("No conditional sync configurations found")
		return nil
	}

	Info("Starting conditional data synchronization")
	
	for tableName, condition := range cfg.TablesSyncConditions {
		Info("Syncing table %s with conditions", tableName)
		
		if err := sc.syncTableWithCondition(tableName, condition); err != nil {
			return NewDatabaseError(fmt.Sprintf("failed to sync table %s", tableName), err)
		}
	}
	
	Info("Conditional data synchronization completed")
	return nil
}

// syncTableWithCondition 根据条件同步单个表
func (sc *SchemaSync) syncTableWithCondition(tableName string, condition TableSyncCondition) error {
	// 获取表结构信息
	sourceFields, err := sc.SourceDb.GetTableFields(tableName)
	if err != nil {
		return NewDatabaseError("failed to get source table fields", err)
	}
	
	destFields, err := sc.DestDb.GetTableFields(tableName)
	if err != nil {
		return NewDatabaseError("failed to get destination table fields", err)
	}
	
	// 确定要同步的字段
	syncFields := condition.SyncFields
	if len(syncFields) == 0 {
		// 如果没有指定字段，使用所有字段
		syncFields = make([]string, 0, len(sourceFields))
		for field := range sourceFields {
			syncFields = append(syncFields, field)
		}
	}
	
	// 验证字段是否存在
	if err := sc.validateSyncFields(tableName, syncFields, sourceFields, destFields); err != nil {
		return err
	}
	
	// 验证主键字段
	if err := sc.validateKeyFields(tableName, condition.KeyFields, sourceFields, destFields); err != nil {
		return err
	}
	
	// 执行删除操作（如果有删除条件）
	if condition.DeleteCondition != "" {
		if err := sc.executeDeleteCondition(tableName, condition.DeleteCondition); err != nil {
			return err
		}
	}
	
	// 执行条件同步
	return sc.executeConditionalSync(tableName, condition, syncFields)
}

// validateSyncFields 验证同步字段
func (sc *SchemaSync) validateSyncFields(tableName string, syncFields []string, sourceFields, destFields map[string]string) error {
	for _, field := range syncFields {
		if _, exists := sourceFields[field]; !exists {
			return NewValidationError(fmt.Sprintf("field %s not found in source table %s", field, tableName), nil)
		}
		if _, exists := destFields[field]; !exists {
			return NewValidationError(fmt.Sprintf("field %s not found in destination table %s", field, tableName), nil)
		}
	}
	return nil
}

// validateKeyFields 验证主键字段
func (sc *SchemaSync) validateKeyFields(tableName string, keyFields []string, sourceFields, destFields map[string]string) error {
	for _, field := range keyFields {
		if _, exists := sourceFields[field]; !exists {
			return NewValidationError(fmt.Sprintf("key field %s not found in source table %s", field, tableName), nil)
		}
		if _, exists := destFields[field]; !exists {
			return NewValidationError(fmt.Sprintf("key field %s not found in destination table %s", field, tableName), nil)
		}
	}
	return nil
}

// executeDeleteCondition 执行删除条件
func (sc *SchemaSync) executeDeleteCondition(tableName, deleteCondition string) error {
	deleteSQL := fmt.Sprintf("DELETE FROM `%s` WHERE %s", tableName, deleteCondition)
	
	Info("Executing delete condition for table %s: %s", tableName, deleteCondition)
	
	if sc.Config.Sync {
		result, err := sc.DestDb.Exec(deleteSQL)
		if err != nil {
			return NewDatabaseError("failed to execute delete condition", err)
		}
		
		rowsAffected, _ := result.RowsAffected()
		Info("Delete condition executed, rows affected: %d", rowsAffected)
	} else {
		Info("Dry run - would execute: %s", deleteSQL)
	}
	
	return nil
}

// executeConditionalSync 执行条件同步
func (sc *SchemaSync) executeConditionalSync(tableName string, condition TableSyncCondition, syncFields []string) error {
	// 构建查询SQL
	fieldsStr := strings.Join(syncFields, "`, `")
	selectSQL := fmt.Sprintf("SELECT `%s` FROM `%s` WHERE %s", fieldsStr, tableName, condition.WhereCondition)
	
	Info("Querying source data: %s", selectSQL)
	
	// 查询源数据
	rows, err := sc.SourceDb.Query(selectSQL)
	if err != nil {
		return NewDatabaseError("failed to query source data", err)
	}
	defer rows.Close()
	
	// 批量处理数据
	batchSize := condition.BatchSize
	if batchSize <= 0 {
		batchSize = 1000
	}
	
	var batch [][]interface{}
	rowCount := 0
	
	for rows.Next() {
		// 创建接收数据的切片
		values := make([]interface{}, len(syncFields))
		valuePointers := make([]interface{}, len(syncFields))
		for i := range values {
			valuePointers[i] = &values[i]
		}
		
		// 扫描数据
		if err := rows.Scan(valuePointers...); err != nil {
			return NewDatabaseError("failed to scan row data", err)
		}
		
		batch = append(batch, values)
		rowCount++
		
		// 达到批量大小时处理数据
		if len(batch) >= batchSize {
			if err := sc.processBatch(tableName, condition, syncFields, batch); err != nil {
				return err
			}
			batch = nil
		}
	}
	
	// 处理剩余数据
	if len(batch) > 0 {
		if err := sc.processBatch(tableName, condition, syncFields, batch); err != nil {
			return err
		}
	}
	
	Info("Conditional sync completed for table %s, processed %d rows", tableName, rowCount)
	return nil
}

// processBatch 处理批量数据
func (sc *SchemaSync) processBatch(tableName string, condition TableSyncCondition, syncFields []string, batch [][]interface{}) error {
	switch condition.SyncMode {
	case "insert_only":
		return sc.processBatchInsert(tableName, syncFields, batch)
	case "update_only":
		return sc.processBatchUpdate(tableName, condition, syncFields, batch)
	case "upsert":
		fallthrough
	default:
		return sc.processBatchUpsert(tableName, condition, syncFields, batch)
	}
}

// processBatchInsert 批量插入
func (sc *SchemaSync) processBatchInsert(tableName string, syncFields []string, batch [][]interface{}) error {
	if len(batch) == 0 {
		return nil
	}
	
	// 构建INSERT语句
	fieldsStr := strings.Join(syncFields, "`, `")
	placeholders := strings.Repeat("?,", len(syncFields))
	placeholders = placeholders[:len(placeholders)-1] // 去掉最后一个逗号
	
	insertSQL := fmt.Sprintf("INSERT IGNORE INTO `%s` (`%s`) VALUES (%s)", tableName, fieldsStr, placeholders)
	
	Info("Batch insert to table %s, rows: %d", tableName, len(batch))
	
	if sc.Config.Sync {
		// 准备批量插入
		stmt, err := sc.DestDb.Prepare(insertSQL)
		if err != nil {
			return NewDatabaseError("failed to prepare insert statement", err)
		}
		defer stmt.Close()
		
		// 执行批量插入
		for _, row := range batch {
			_, err := stmt.Exec(row...)
			if err != nil {
				Warn("Failed to insert row: %v, error: %v", row, err)
			}
		}
	} else {
		Info("Dry run - would insert %d rows into table %s", len(batch), tableName)
	}
	
	return nil
}

// processBatchUpdate 批量更新
func (sc *SchemaSync) processBatchUpdate(tableName string, condition TableSyncCondition, syncFields []string, batch [][]interface{}) error {
	if len(batch) == 0 {
		return nil
	}
	
	// 构建UPDATE语句
	var setParts []string
	keyFieldsMap := make(map[string]bool)
	
	for _, keyField := range condition.KeyFields {
		keyFieldsMap[keyField] = true
	}
	
	for _, field := range syncFields {
		if !keyFieldsMap[field] {
			setParts = append(setParts, fmt.Sprintf("`%s` = ?", field))
		}
	}
	
	var whereParts []string
	for _, keyField := range condition.KeyFields {
		whereParts = append(whereParts, fmt.Sprintf("`%s` = ?", keyField))
	}
	
	updateSQL := fmt.Sprintf("UPDATE `%s` SET %s WHERE %s", tableName, strings.Join(setParts, ", "), strings.Join(whereParts, " AND "))
	
	Info("Batch update to table %s, rows: %d", tableName, len(batch))
	
	if sc.Config.Sync {
		stmt, err := sc.DestDb.Prepare(updateSQL)
		if err != nil {
			return NewDatabaseError("failed to prepare update statement", err)
		}
		defer stmt.Close()
		
		// 为每一行构建参数
		for _, row := range batch {
			var params []interface{}
			fieldIndexMap := make(map[string]int)
			
			for i, field := range syncFields {
				fieldIndexMap[field] = i
			}
			
			// 添加SET部分的参数
			for _, field := range syncFields {
				if !keyFieldsMap[field] {
					params = append(params, row[fieldIndexMap[field]])
				}
			}
			
			// 添加WHERE部分的参数
			for _, keyField := range condition.KeyFields {
				params = append(params, row[fieldIndexMap[keyField]])
			}
			
			_, err := stmt.Exec(params...)
			if err != nil {
				Warn("Failed to update row: %v, error: %v", row, err)
			}
		}
	} else {
		Info("Dry run - would update %d rows in table %s", len(batch), tableName)
	}
	
	return nil
}

// processBatchUpsert 批量插入或更新
func (sc *SchemaSync) processBatchUpsert(tableName string, condition TableSyncCondition, syncFields []string, batch [][]interface{}) error {
	if len(batch) == 0 {
		return nil
	}
	
	// 构建ON DUPLICATE KEY UPDATE语句
	fieldsStr := strings.Join(syncFields, "`, `")
	placeholders := strings.Repeat("?,", len(syncFields))
	placeholders = placeholders[:len(placeholders)-1]
	
	var updateParts []string
	keyFieldsMap := make(map[string]bool)
	
	for _, keyField := range condition.KeyFields {
		keyFieldsMap[keyField] = true
	}
	
	for _, field := range syncFields {
		if !keyFieldsMap[field] {
			updateParts = append(updateParts, fmt.Sprintf("`%s` = VALUES(`%s`)", field, field))
		}
	}
	
	upsertSQL := fmt.Sprintf("INSERT INTO `%s` (`%s`) VALUES (%s) ON DUPLICATE KEY UPDATE %s", 
		tableName, fieldsStr, placeholders, strings.Join(updateParts, ", "))
	
	Info("Batch upsert to table %s, rows: %d", tableName, len(batch))
	
	if sc.Config.Sync {
		stmt, err := sc.DestDb.Prepare(upsertSQL)
		if err != nil {
			return NewDatabaseError("failed to prepare upsert statement", err)
		}
		defer stmt.Close()
		
		for _, row := range batch {
			_, err := stmt.Exec(row...)
			if err != nil {
				Warn("Failed to upsert row: %v, error: %v", row, err)
			}
		}
	} else {
		Info("Dry run - would upsert %d rows into table %s", len(batch), tableName)
	}
	
	return nil
}

// CheckSchemaDiff 检查schema差异
func CheckSchemaDiff(cfg *Config) error {
	sc, err := NewSchemaSync(cfg)
	if err != nil {
		return err
	}
	
	allTables, err := sc.SourceDb.GetTableNames()
	if err != nil {
		return NewDatabaseError("failed to get source table names", err)
	}
	
	dstTables, err := sc.DestDb.GetTableNames()
	if err != nil {
		return NewDatabaseError("failed to get destination table names", err)
	}
	
	// 检查新表
	newTables := []string{}
	for _, table := range allTables {
		if !inStringSlice(table, dstTables) {
			newTables = append(newTables, table)
		}
	}
	
	if len(newTables) > 0 {
		fmt.Printf("# New tables in source: %s\n", strings.Join(newTables, ", "))
	}
	
	// 检查表结构差异
	for _, table := range allTables {
		if !cfg.CheckMatchTables(table) {
			continue
		}
		
		if inStringSlice(table, dstTables) {
			alterData, err := sc.getAlterDataByTable(table, cfg)
			if err != nil {
				return NewDatabaseError(fmt.Sprintf("failed to get alter data for table %s", table), err)
			}
			
			if len(alterData.SQL) > 0 {
				fmt.Printf("# Table %s needs schema changes:\n", table)
				for _, sql := range alterData.SQL {
					fmt.Printf("%s\n", sql)
				}
			}
		}
	}
	
	return nil
}

// CompareSqlResult 比较SQL在两个数据库中的执行结果
func CompareSqlResult(cfg *Config, sql string) error {
	sc, err := NewSchemaSync(cfg)
	if err != nil {
		return err
	}
	
	// 在源数据库执行SQL
	sourceRows, err := sc.SourceDb.Query(sql)
	if err != nil {
		return NewDatabaseError("failed to execute SQL on source database", err)
	}
	defer sourceRows.Close()
	
	// 在目标数据库执行SQL
	destRows, err := sc.DestDb.Query(sql)
	if err != nil {
		return NewDatabaseError("failed to execute SQL on destination database", err)
	}
	defer destRows.Close()
	
	// 获取列信息
	columns, err := sourceRows.Columns()
	if err != nil {
		return NewDatabaseError("failed to get columns", err)
	}
	
	fmt.Printf("# Comparing SQL: %s\n", sql)
	fmt.Printf("# Columns: %s\n", strings.Join(columns, ", "))
	
	// 比较结果
	sourceResults := [][]interface{}{}
	destResults := [][]interface{}{}
	
	// 读取源数据库结果
	for sourceRows.Next() {
		values := make([]interface{}, len(columns))
		valuePointers := make([]interface{}, len(columns))
		for i := range values {
			valuePointers[i] = &values[i]
		}
		
		if err := sourceRows.Scan(valuePointers...); err != nil {
			return NewDatabaseError("failed to scan source results", err)
		}
		sourceResults = append(sourceResults, values)
	}
	
	// 读取目标数据库结果
	for destRows.Next() {
		values := make([]interface{}, len(columns))
		valuePointers := make([]interface{}, len(columns))
		for i := range values {
			valuePointers[i] = &values[i]
		}
		
		if err := destRows.Scan(valuePointers...); err != nil {
			return NewDatabaseError("failed to scan destination results", err)
		}
		destResults = append(destResults, values)
	}
	
	// 比较结果
	fmt.Printf("# Source results: %d rows\n", len(sourceResults))
	fmt.Printf("# Destination results: %d rows\n", len(destResults))
	
	if len(sourceResults) != len(destResults) {
		fmt.Printf("# WARNING: Result count mismatch!\n")
	}
	
	// 显示前几行结果
	maxRows := 10
	if len(sourceResults) < maxRows {
		maxRows = len(sourceResults)
	}
	
	fmt.Printf("# Source results (first %d rows):\n", maxRows)
	for i := 0; i < maxRows; i++ {
		fmt.Printf("  %v\n", sourceResults[i])
	}
	
	if len(destResults) < maxRows {
		maxRows = len(destResults)
	}
	
	fmt.Printf("# Destination results (first %d rows):\n", maxRows)
	for i := 0; i < maxRows; i++ {
		fmt.Printf("  %v\n", destResults[i])
	}
	
	return nil
}
